package cn.panshi.mongocdc;

import java.io.Serializable;
import java.lang.reflect.Method;
import java.util.Date;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.bson.Document;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.messaging.Message;
import org.springframework.data.mongodb.core.messaging.MessageListener;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.data.redis.core.RedisTemplate;

import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.client.model.changestream.OperationType;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@SuppressWarnings({ "rawtypes" })
@Data
public class MongoCdcBizDispatchListener implements MessageListener<ChangeStreamDocument<Document>, Document> {

	private MongoTemplate mongoTemplate;
	private RedisTemplate redisTemplate;

	private IMongoCdcBizRouter mongoCdcBizRouter;
	private IMongoCdcTtlTaskRouter mongoCdcTtlTaskRouter;

	@Override
	public void onMessage(Message<ChangeStreamDocument<Document>, Document> message) {
		String collectionName = message.getProperties().getCollectionName();
		if (StringUtils.equals(collectionName, CdcOptLogDoc.TABLE)) {
			return;
		}

		OperationType operationType = message.getRaw().getOperationType();

		if (CdcTtlTaskDoc.TABLE.equals(collectionName)) {
			this.processCdc4TtlTask(collectionName, operationType, message);
		} else {
			this.processCdc(collectionName, operationType, message);
		}

	}

	private void processCdc4TtlTask(String collectionName, OperationType operationType,
			Message<ChangeStreamDocument<Document>, Document> message) {
		try {
			new AbsCdcTemplate(redisTemplate, mongoTemplate) {

				@Override
				protected boolean supportsOperation(String collectionName, OperationType operationType) {
					return CdcTtlTaskDoc.TABLE.equals(collectionName) && operationType == OperationType.DELETE;
				}

				@Override
				protected void executeCdcBiz(String bizId, String resumeToken,
						Message<ChangeStreamDocument<Document>, Document> message, String collectionName,
						OperationType operationType) throws Exception {
					CdcTtlTaskDoc task = findTaskByTriggerId(bizId);
					if (task == null) {
						log.warn("No ttl task found by {}", bizId);
						return;
					}

					if ((task.getExecutedTime() != null) || task.getUnscheduleTime() != null) {
						log.debug("################task {} executed {} {},unscheduleTime {}", task.getName(),
								task.getExecutionMemo(), task.getExecutedTime(), task.getUnscheduleTime());
						return;
					}

					try {
						Serializable result = mongoCdcTtlTaskRouter.execute(task);
						updateTaskResult(task.getId(), result);
					} catch (Exception e) {
						log.error(e.getMessage(), e);
						updateTaskResult(task.getId(), ReflectionUtils.calcInvokeStackTrace(e));
					}

				}

				private void updateTaskResult(String id, Serializable result) {
					Query query = new Query(Criteria.where("_id").is(id));
					Update update = new Update().set("executedTime", new Date()).set("executionMemo", result);
					mongoTemplate.updateFirst(query, update, CdcTtlTaskDoc.TABLE);
				}

				private CdcTtlTaskDoc findTaskByTriggerId(String triggerId) {
					Query query = new Query(Criteria.where("triggerId").is(triggerId));
					List<CdcTtlTaskDoc> list = mongoTemplate.find(query, CdcTtlTaskDoc.class, CdcTtlTaskDoc.TABLE);

					return list.isEmpty() ? null : list.get(0);
				}
			}.process(collectionName, operationType, message);
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		}

	}

	@SuppressWarnings("unchecked")
	private void processCdc(String collectionName, OperationType operationType,
			Message<ChangeStreamDocument<Document>, Document> message) {
		try {
			new AbsCdcTemplate4CreateUpdateRouter(redisTemplate, mongoTemplate) {

				@Override
				protected List<Method> matchBizMethodByCdc(Enum cdcType) {
					return mongoCdcBizRouter.matchBizMethodByCdc(cdcType);
				}

				@Override
				protected List<Enum> parseCdcType(String collectionName, OperationType operationType) {
					return mongoCdcBizRouter.parse(collectionName, operationType);
				}

				@Override
				protected void invokeBizRouter(Method matchedRouterFunc, String bizId, String resumeToken,
						Message<ChangeStreamDocument<Document>, Document> message) throws Exception {
					matchedRouterFunc.invoke(mongoCdcBizRouter, bizId, resumeToken, message);
				}
			}.process(collectionName, operationType, message);
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		}

	}
}